package com.cx.mall.demo.handler;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;

import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

@Configuration
public class ConsumerHandler {
    @Autowired
    private RedisTemplate redisTemplate;

    @Bean
    public Consumer<Message<String>> testInput() {
        return message -> {
            String payload = message.getPayload();
            Map<String, Object> headers = message.getHeaders();
            System.out.println("test收到消息: " + payload + ", 队列号: " + headers.get("ROCKET_MQ_QUEUE_ID"));
        };
    }

    @Bean
    public Consumer<Message<String>> test1Input() {
        return message -> {
            String payload = message.getPayload();
            Map<String, Object> headers = message.getHeaders();
            System.out.println("test1收到消息: " + payload + ", 队列号: " + headers.get("ROCKET_MQ_QUEUE_ID"));
        };
    }

    @Bean
    public Consumer<Message<String>> test2Input() {
        return message -> {
            String payload = message.getPayload();
            Map<String, Object> headers = message.getHeaders();
            System.out.println("test2收到消息: " + payload + ", 队列号: " + headers.get("ROCKET_MQ_QUEUE_ID"));
        };
    }

    //通过redis的幂等性来解决消息重复发送
    @Bean
    public Consumer<Message<String>> test3Input() {
        return message -> {
            String payload = message.getPayload();
            Map<String, Object> headers = message.getHeaders();
            String key = "QUE_" + (String) headers.get("ROCKET_MQ_MESSAGE_ID");
            if (redisTemplate.opsForValue().setIfAbsent(key, "1", 30, TimeUnit.SECONDS)) {
                System.out.println("test3收到消息: " + payload);
            }
            throw new RuntimeException("test3异常");
        };
    }

}
